package com.raft.core.log.entry;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.Serializable;
import java.util.Collection;
import java.util.Set;
import java.util.stream.Collectors;

import com.alibaba.com.caucho.hessian.io.HessianInput;
import com.raft.core.node.NodeEndpoint;

public class EntryFactory {

    public Entry create(int kind, int index, int term, byte[] commandBytes) {
        switch (kind) {
            case Entry.KIND_NO_OP:
                return new NoOpEntry(index, term);
            case Entry.KIND_GENERAL:
                return new GeneralEntry(index, term, commandBytes);
            case Entry.KIND_ADD_NODE:
            	AddNodeEntry addNodeCommand = decode(commandBytes, AddNodeEntry.class);
                return new AddNodeEntry(index, term, asNodeEndpoints(addNodeCommand.getNodeEndpoints()), asNodeEndpoint(addNodeCommand.getNewNodeEndpoint()));
            case Entry.KIND_REMOVE_NODE:
            	RemoveNodeEntry removeNodeCommand = decode(commandBytes, RemoveNodeEntry.class);
                return new RemoveNodeEntry(index, term, asNodeEndpoints(removeNodeCommand.getNodeEndpoints()), removeNodeCommand.getNodeToRemove());
            default:
                throw new IllegalArgumentException("unexpected entry kind " + kind);
            }
    }

    private Set<NodeEndpoint> asNodeEndpoints(Collection<NodeEndpoint> protoNodeEndpoints) {
        return protoNodeEndpoints.stream().map(this::asNodeEndpoint).collect(Collectors.toSet());
    }

    private NodeEndpoint asNodeEndpoint(NodeEndpoint protoNodeEndpoint) {
        return new NodeEndpoint(protoNodeEndpoint.getId(), protoNodeEndpoint.getHost(), protoNodeEndpoint.getPort());
    }

    private <T extends Serializable> T decode(byte[] bytes, Class<T> clazz) {
        if (bytes == null) {
            return null;
        }
        // 1、将字节数组转换成字节输入流
        ByteArrayInputStream bis = new ByteArrayInputStream(bytes);

        // step 1. 定义外部序列化工厂
        //ExtSerializerFactory extSerializerFactory = new ExtSerializerFactory();
        //extSerializerFactory.addSerializer(java.time.OffsetDateTime.class, new OffsetDateTimeRedisSerializer());
        //extSerializerFactory.addDeserializer(java.time.OffsetDateTime.class, new OffsetDateTimeRedisDeserializer());
        // step 2. 序列化工厂
        //SerializerFactory serializerFactory = new SerializerFactory();
        //serializerFactory.addFactory(extSerializerFactory);
        HessianInput hessianInput = new HessianInput(bis);
        //hessianInput.setSerializerFactory(serializerFactory);
        Object object = null;

        try {
            object = hessianInput.readObject();
        } catch (IOException e) {
        	e.printStackTrace();
        } catch (Exception e) {
        	e.printStackTrace();
        }

        return (T) object;
    }

}
